package com.lagou.kafka.producer;

import com.lagou.kafka.partitioner.MyPartitioner;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description: 发送者
 * @date 2022/8/18 20:42
 */
public class MyProducer {
    public static void main(String[] args) {

        Map<String, Object> producerMap = new HashMap<>();
        producerMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.106.130:9092");
        producerMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerMap.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
        // 定义自定义的分区器
        KafkaProducer<String,String> producer = new KafkaProducer<>(producerMap);

        ProducerRecord<String,String> record = new ProducerRecord<>(
                "tp_user_01", "1",
                "欧阳钊"
        );

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if(exception != null){
                    throw new RuntimeException("发送失败...");
                }else{
                    System.out.println(metadata.topic());
                    System.out.println(metadata.partition());
                    System.out.println(metadata.offset());
                }
            }
        });
        // 关闭发送的流
        producer.close();
    }

}
